package com.gitee.wsl.flow.coroutines

import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.launch

internal fun <T> Flow<T>.collectIntoChannel(scope: CoroutineScope): Channel<T> {
    val output = Channel<T>(UNLIMITED)
    val job = scope.launch(start = CoroutineStart.UNDISPATCHED) {
        try {
            collect { output.trySend(it) }
            output.close()
        } catch (e: Throwable) {
            output.close(e)
        }
    }

    return object : Channel<T> by output {
        override fun cancel(cause: CancellationException?) {
            job.cancel()
            output.close(cause)
        }

        override fun close(cause: Throwable?): Boolean {
            job.cancel()
            return output.close(cause)
        }
    }
}
